package com.fbs.job.core.executor;

import com.fbs.job.core.biz.AdminBiz;
import com.fbs.job.core.biz.client.AdminBizClient;
import com.fbs.job.core.handler.IJobHandler;
import com.fbs.job.core.handler.annotation.FbsJob;
import com.fbs.job.core.handler.impl.MethodJobHandler;
import com.fbs.job.core.log.FbsJobFileAppender;
import com.fbs.job.core.server.EmbedServer;
import com.fbs.job.core.thread.JobLogFileCleanThread;
import com.fbs.job.core.thread.JobThread;
import com.fbs.job.core.thread.TriggerCallbackThread;
import com.fbs.job.core.util.IpUtil;
import com.fbs.job.core.util.PortUtil;
import org.apache.commons.lang3.StringUtils;
import org.slf4j.Logger;
import org.slf4j.LoggerFactory;

import java.lang.reflect.Method;
import java.util.ArrayList;
import java.util.List;
import java.util.Map;
import java.util.concurrent.ConcurrentHashMap;
import java.util.concurrent.ConcurrentMap;

public class FbsJobExecutor {

    private static final Logger logger = LoggerFactory.getLogger(FbsJobExecutor.class);
    private String adminAddresses;
    private String accessToken;
    private String appName;
    private String address;
    private String ip;
    private int port;
    private String logPath;
    private int logRetentionDays;
    public void setAdminAddresses(String adminAddresses) {
        this.adminAddresses = adminAddresses;
    }
    public void setAccessToken(String accessToken) {
        this.accessToken = accessToken;
    }
    public void setAppname(String appName) {
        this.appName = appName;
    }
    public void setAddress(String address) {
        this.address = address;
    }
    public void setIp(String ip) {
        this.ip = ip;
    }
    public void setPort(int port) {
        this.port = port;
    }
    public void setLogPath(String logPath) {
        this.logPath = logPath;
    }
    public void setLogRetentionDays(int logRetentionDays) {
        this.logRetentionDays = logRetentionDays;
    }
    // ---------------------- executor-server (rpc provider) ----------------------
    private EmbedServer embedServer = null;

     public void start(){


         // init logpath
         FbsJobFileAppender.initLogPath(logPath);

         initAdminBizList(adminAddresses, accessToken);
         // init JobLogFileCleanThread
         JobLogFileCleanThread.getInstance().start(logRetentionDays);
//         //初始化回调线程
         // init TriggerCallbackThread
         TriggerCallbackThread.getInstance().start();
         //初始化netty
         initEmbedServer(address, ip, port, appName, accessToken);
     }

    public void destroy(){
        stopEmbedServer();
        // destroy jobThreadRepository
        if (jobThreadRepository.size() > 0) {
            for (Map.Entry<Integer, JobThread> item: jobThreadRepository.entrySet()) {
                JobThread oldJobThread = removeJobThread(item.getKey(), "web container destroy and kill the job.");
                // wait for job thread push result to callback queue
                if (oldJobThread != null) {
                    try {
                        oldJobThread.join();
                    } catch (InterruptedException e) {
                        logger.error(">>>>>>>>>>> fbs-job, JobThread destroy(join) error, jobId:{}", item.getKey(), e);
                    }
                }
            }
            jobThreadRepository.clear();
        }
        jobHandlerRepository.clear();
        //         //初始化回调线程
        // init TriggerCallbackThread
        TriggerCallbackThread.getInstance().toStop();
        //初始化netty

        JobLogFileCleanThread.getInstance().toStop();
    }

    private void stopEmbedServer() {
        // stop provider factory
        if (embedServer != null) {
            try {
                embedServer.stop();
            } catch (Exception e) {
                logger.error(e.getMessage(), e);
            }
        }
    }
    private void initEmbedServer(String address, String ip, int port, String appName, String accessToken) {
        if (port<=0){
            port = PortUtil.findCanUsePort(9999);
        }
        if (StringUtils.isBlank(ip)){
            ip= IpUtil.getIp();
        }
        if (StringUtils.isBlank(address)){
            String ip_port_address = IpUtil.getIpPort(ip, port);   // registry-address：default use address to registry , otherwise use ip:port if address is null
            address =     "http://"+ip_port_address+"/";
        }

        // accessToken
        if (accessToken==null || accessToken.trim().length()==0) {
            logger.warn(">>>>>>>>>>> fbs-job accessToken is empty. To ensure system security, please set the accessToken.");
        }

        // start
        embedServer = new EmbedServer();
        embedServer.start(address, port, appName, accessToken);
    }
    public static List<AdminBiz> getAdminBizList(){
        return adminBizList;
    }
    // ---------------------- admin-client (rpc invoker) ----------------------
    private static List<AdminBiz> adminBizList;
    private void initAdminBizList(String adminAddresses, String accessToken) {
        if (StringUtils.isBlank(adminAddresses)){
            return;
        }

        for (String address : adminAddresses.trim().split(",")) {
            if (StringUtils.isNotBlank(address)){
                AdminBiz adminBiz = new AdminBizClient(address.trim(),accessToken);
                if (adminBizList == null) {
                    adminBizList = new ArrayList<>();
                }
                adminBizList.add(adminBiz);
            }
        }
    }




    //----------------------------------------------客户端 执行器注册相关方法---------------------------------------------------
    // ---------------------- job handler repository ----------------------
    private static ConcurrentMap<String, IJobHandler> jobHandlerRepository = new ConcurrentHashMap<String, IJobHandler>();
    public static IJobHandler loadJobHandler(String name){
        return jobHandlerRepository.get(name);
    }
    public static IJobHandler registerJobHandler(String name, IJobHandler jobHandler){
        logger.info(">>>>>>>>>>> fbs-job register jobhandler success, name:{}, jobHandler:{}", name, jobHandler);
        return jobHandlerRepository.put(name, jobHandler);
    }

    public void registerJobHandler(FbsJob fbsJob, Object bean, Method method){
        if (fbsJob == null) {
            return;
        }
        String name = fbsJob.value();
        Class<?> clazz = bean.getClass();
        String methodName = method.getName();
        if (StringUtils.isBlank(name)) {
            throw new RuntimeException("fbs-job method-jobhandler name invalid, for[" + clazz + "#" + methodName + "] .");
        }
        if (loadJobHandler(name) != null) {
            throw new RuntimeException("fbs-job jobhandler[" + name + "] naming conflicts.");
        }
        method.setAccessible(true);
        // init and destroy
        Method initMethod = null;
        Method destroyMethod = null;
        if (fbsJob.init().trim().length() > 0) {
            try {
                initMethod = clazz.getDeclaredMethod(fbsJob.init());
                initMethod.setAccessible(true);
            } catch (NoSuchMethodException e) {
                throw new RuntimeException("fbs-job method-jobhandler initMethod invalid, for[" + clazz + "#" + methodName + "] .");
            }
        }
        if (fbsJob.destroy().trim().length() > 0) {
            try {
                destroyMethod = clazz.getDeclaredMethod(fbsJob.destroy());
                destroyMethod.setAccessible(true);
            } catch (NoSuchMethodException e) {
                throw new RuntimeException("fbs-job method-jobhandler destroyMethod invalid, for[" + clazz + "#" + methodName + "] .");
            }
        }
        // registry jobhandler
        registerJobHandler(name, new MethodJobHandler(bean, method, initMethod, destroyMethod));
    }

    // ---------------------- job thread repository ----------------------
    private static ConcurrentMap<Integer, JobThread> jobThreadRepository = new ConcurrentHashMap<>();
    public static JobThread registerJobThread(int jobId, IJobHandler handler, String removeOldReason){
        JobThread newJobThread = new JobThread(jobId, handler);
        newJobThread.start();
        logger.info(">>>>>>>>>>> fbs-job regist JobThread success, jobId:{}, handler:{}", new Object[]{jobId, handler});

        JobThread oldJobThread = jobThreadRepository.put(jobId, newJobThread);	// putIfAbsent | oh my god, map's put method return the old value!!!
        if (oldJobThread != null) {
            oldJobThread.toStop(removeOldReason);
            oldJobThread.interrupt();
        }

        return newJobThread;
    }
    public static JobThread removeJobThread(int jobId, String removeOldReason){
        JobThread oldJobThread = jobThreadRepository.remove(jobId);
        if (oldJobThread != null) {
            oldJobThread.toStop(removeOldReason);
            oldJobThread.interrupt();

            return oldJobThread;
        }
        return null;
    }

    public static JobThread loadJobThread(int jobId){
        return jobThreadRepository.get(jobId);
    }

}
